package top.easyblog.core.support.queue;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.data.redis.core.RedisTemplate;
import top.easyblog.common.enums.CompressionCodeStatus;
import top.easyblog.core.config.CustomConfiguration;
import top.easyblog.core.dao.auto.model.CompressionCode;
import top.easyblog.core.service.CompressCodeService;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author: frank.huang
 * @date: 2023-12-31 14:39
 */
@Slf4j
public class CompressionCodeBlockingQueue {


    private final CompressCodeService compressCodeService;

    //缓存压缩码到内存中
    private final RedisBlockingQueue<Object> queue;

    //默认加载数量
    private final static int DEFAULT_LOAD_NUM = 100000;

    private final static int COMPRESSION_CODE_ADD_THRESHOLD = 5000000;


    public CompressionCodeBlockingQueue(CustomConfiguration customConfiguration,
                                        CompressCodeService compressCodeService,
                                        RedisTemplate<String, Object> redisTemplate) {
        this.compressCodeService = compressCodeService;
        this.queue = new RedisBlockingQueue<>(customConfiguration.getCompressionCodeQueueKey(),
                redisTemplate, customConfiguration.getCompressionCodeQueueSize());
        initQueue();
    }


    public void initQueue() {
        if (!queue.isEmpty()) {
            log.info("Compression code queue is not empty,end init!");
            return;
        }

        log.info("Try init compression code queue...");
        List<CompressionCode> compressCodeList = compressCodeService.getCompressCodeList(100000, NumberUtils.INTEGER_ZERO);
        if (CollectionUtils.isNotEmpty(compressCodeList)) {
            List<String> codes = compressCodeList.stream().filter(Objects::nonNull).map(CompressionCode::getCompressionCode)
                    .collect(Collectors.toList());
            queue.addAll(codes.toArray(new String[]{}));
            compressCodeService.batchUpdateStatus(codes, CompressionCodeStatus.OCCUPIED.getValue());
        }
        log.info("Init compression code queue!");
    }

    public void batchOffer() {
        int size = queue.size();
        log.info("Current compression queue size is {}", size);
        if (size > COMPRESSION_CODE_ADD_THRESHOLD) {
            log.info("Queue compression code not touch threshold,end add task.");
            return;
        }

        List<CompressionCode> compressCodeList = compressCodeService.getCompressCodeList(DEFAULT_LOAD_NUM, NumberUtils.INTEGER_ZERO);
        if(CollectionUtils.isEmpty(compressCodeList)){
            log.info("DB available compression is empty.");
            return;
        }

        List<String> codes = compressCodeList.stream().map(CompressionCode::getCompressionCode)
                .parallel().collect(Collectors.toList());
        queue.addAll(codes.toArray(new String[]{}));
        compressCodeService.batchUpdateStatus(codes, CompressionCodeStatus.OCCUPIED.getValue());
    }


    public String poll() {
        return (String) queue.poll();
    }


}
